package org.example
import org.apache.spark.sql.SparkSession
import java.util.regex.Pattern
object data1_movies {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    val filePath = "src/main/resources/"
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")
       println("电影总数：" + moviesRDD.count() + "\n" + "用户总数" + usersRDD.count())
       println("评分总数" + ratingsRDD.count() + "\n" + "职业数：" + occupationsRDD.count())
//    val num = ratingsRDD.map(_.split("::"))
//       .map(user => (user(1),user(0)))
//       .filter(_._2.equals("3088"))
//       println("用户Id为3088的评分电影总数为：" + num.count())
//    val moviesInfo = moviesRDD.map(_.split("::"))
//       .map(movies => (movies(0),(movies(1),movies(2))))
//    val res = num.join(moviesInfo)
//       .map(item => (item._1,item._2._1,item._2._2._1,item._2._2._2))
//       println("电影详情是：\n")
//       res.take(3).foreach(println)

//分析二：找出电影评分最高的前10部电影名，观影人数最多的前10部电影
    val rating = ratingsRDD.map(x => x.split("::")).map {
      x => {
        (x(0), x(1), x(2))
      }
    }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
    rating.map(x => (x._2, (x._3.toDouble,1)))
      .reduceByKey((x,y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._2._1 / x._2._2,x._1))
      .sortByKey(false)
      .take(5)
      .foreach(println)
    println("按平均分取前 10 部电影输出详情：(平均分,(movieId,Title,Genres,总分,总次数))")
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
      }).sortByKey(false)
      .take(5)
      .foreach(println)

    println("观看人数最多的前10部电影")
    val watchViewsInfo = rating.map(x => {
        (x._2, 1)
    }).reduceByKey((x,y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5)
    watchViewsInfo.foreach(println(_))
    println("==================>")

    rating.map(x => (x._2, 1))
      .reduceByKey((x,y) => {
        (x + y)
      })
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)

    println("详情的输出( 观影人数,电影编号)")
    moviesInfo.join(ratingsInfo).map(x => {
      (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
    }).sortByKey(false)
      .take(10)
      .foreach(println)
    //分析三：统计不同电影类型的数量
    val movieStyle = moviesRDD.map(_.split("::"))
      .map(tp => (tp(0),tp(2)))
      .flatMapValues(sp => sp.split("\\|"))
      .map(p => (p._2,1))
      .reduceByKey((x,y) => x + y)
      .filter(_._1.equals("Comedy"))
    movieStyle.foreach(println)
    //分析四：统计每年度生产电影总数
//    val pattern = Pattern.compile(" (.*)(\\(\\d{4}\\))")
//    val movieYear = moviesRDD.map(_.split("::"))
//      .map(x => (x(1),1))
//      .map((x => {
//        var name = ""
//        var year = ""
//        var matcher = pattern.matcher(x._1)
//        if (matcher.find()) {
//          name = matcher.group(1)
//          year = matcher.group(2)
//          year = year.substring(1, year.length - 1)
//        }
//        if (year == "") {
//          (-1,1)
//        }else{
//          (year.toInt,1)
//        }
//      }))
//      .reduceByKey((x,y) => x + y)
//      .sortByKey()
//      .collect()
//    movieYear.take(10).foreach(println)
    //分析五：2000年度生产的电影总数
//    val pattern = Pattern.compile("repare'\\['sp\\]\\[t\\]\\[t\\]\\[d\\]\\[s\\]")
//    val movies2000Count = moviesRDD.map(_.split("repare='utf-8'"))
//      .map(x => (x(1),1))
//      .map(x =>  {
//        var name = ""
//        var year = "2000"
//        val matcher = pattern.matcher(x._1)
//        if (matcher.find()){
//          name = matcher.group(1)
//          year = matcher.group(2)
//          year = year.substring(0, year.length - 1)
//        }
//          .filter(_._1 == "2000")
//          .reduceBykey((x, y) => x + y)
//          .collect()
//        movies2000Count.foreach(println)
//      })
//分析五：2000年度生产的电影总数
    val count2000 = moviesRDD
  .map(_.split("::")(1))
  .filter(title => title.contains("(2000)"))
  .count()
    println(s"2000年生产的电影总数量：$count2000")
  }
}
